package com.jy.datapipeline.export.service.impl;

import com.jy.datapipeline.common.config.DBMetaDataConfig;
import com.jy.datapipeline.common.constants.ExpConstants;
import com.jy.datapipeline.common.exception.GlobalException;
import com.jy.datapipeline.export.dao.ExpTaskDao;
import com.jy.datapipeline.export.dao.repo.DataSourceInfoDaoRepo;
import com.jy.datapipeline.export.dao.repo.DataStoreInfoDaoRepo;
import com.jy.datapipeline.export.dao.repo.ExpRuleDaoRepo;
import com.jy.datapipeline.export.dao.repo.ExpRuleDetailDaoRepo;
import com.jy.datapipeline.export.dao.ExpTaskExecuteDetailLogDao;
import com.jy.datapipeline.export.dao.repo.ExpTaskExecuteDetailLogDaoRepo;
import com.jy.datapipeline.export.dao.ExpTaskExecuteLogDao;
import com.jy.datapipeline.export.dao.repo.ExpTaskExecuteLogDaoRepo;
import com.jy.datapipeline.export.entity.*;
import com.jy.datapipeline.export.common.enums.impl.SplitKeyExecutorEnums;
import com.jy.datapipeline.export.service.ExpRuleDetailService;
import com.jy.datapipeline.export.splitkey.adapter.SplitKeyAdapter;
import com.jy.datapipeline.export.splitkey.split.impl.AccurateSplit;
import com.jy.datapipeline.export.splitkey.split.impl.RoughSplit;
import com.jy.datapipeline.export.splitkey.handler.SplitKeyTaskHandler;
import com.jy.datapipeline.export.splitkey.adapter.SplitKeyExecutor;
import com.jy.datapipeline.export.service.DataDictionaryService;
import com.jy.datapipeline.export.service.ExpTaskExecuteLogService;
import com.jy.datapipeline.export.task.TaskExecuteCallable;

import java.util.List;
import java.util.Map;
import javax.annotation.Resource;
import javax.sql.DataSource;
import org.quartz.Scheduler;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.transaction.annotation.Transactional;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

@Slf4j
@Service
@Transactional
public class ExpTaskExecuteLogServiceImpl implements ExpTaskExecuteLogService {
    @Autowired
    private ExpTaskExecuteLogDao expTaskExecuteLogDao;
    @Autowired
    private ExpTaskExecuteLogDaoRepo expTaskExecuteLogDaoRepo;
    @Autowired
    private ExpRuleDetailDaoRepo expRuleDetailDaoRepo;
    @Autowired
    private ExpTaskExecuteDetailLogDao expTaskExecuteDetailLogDao;
    @Qualifier("asyncTaskExecute")
    @Autowired
    public ThreadPoolTaskExecutor asyncTaskExecute;
    @Autowired
    private ExpRuleDaoRepo expRuleDaoRepo;
    @Autowired
    private DataStoreInfoDaoRepo dataStoreInfoDaoRepo;
    @Autowired
    private ExpTaskExecuteDetailLogDaoRepo expTaskExecuteDetailLogDaoRepo;
    @Autowired
    private DataSourceInfoDaoRepo dataSourceInfoDaoRepo;

    @Resource
    private DataDictionaryService dataDictionaryService;

    @Autowired
    private ExpTaskDao expTaskDao;
    @Value("${config.localStorePath}")
    public   String localStorePath;
    @Value("${config.fetchSize}")
    public  int fetchSize;
    @Value("${config.printOutSize}")
    public  int printOutSize;

    @Autowired
    private Scheduler scheduler;

    @Resource
    private RoughSplit roughSplit;

    @Resource
    private AccurateSplit accurateSplit;

    @Resource
    private ExpRuleDetailService expruleDetailService;

    @Override
    public List<ExpTaskExecuteLogModel> findByTaskIdAndStatus(String taskId, int status) {
        return this.expTaskExecuteLogDaoRepo.findByTaskIdAndStatus(taskId,status);
    }

    @Override
    public List<ExpTaskExecuteLogModel> findByTaskId(String taskId) {
        return this.expTaskExecuteLogDaoRepo.findByTaskId(taskId);
    }

    @Override
    public List<Map<String, Object>> findExpTaskExecuteLogByList(String taskId, int iDisplayStart,
        int iDisplayLength,String status) {
        return this.expTaskExecuteLogDao.findExpTaskExecuteLogByList(taskId,iDisplayStart,iDisplayLength, status);
    }

    @Override
    public int findExpTaskExecuteLogByCount(String taskId,String stataus) {
        return this.expTaskExecuteLogDao.findExpTaskExecuteLogByCount(taskId,stataus);
    }

    @Override
    public boolean deleteByTaskId(String taskId) {
        return this.expTaskExecuteLogDao.deleteByTaskId(taskId);
    }

    @Override
    public ExpTaskExecuteLogModel startTaskExeucte(ExpTaskModel etm) {
        String executeLogId="task_exec_"+System.nanoTime();
        ExpTaskExecuteLogModel etelm=new ExpTaskExecuteLogModel();
        try{
            etelm.setTaskExecuteLogId(executeLogId);
            etelm.setTaskId(etm.getTaskId());
            etelm.setStatus(1);
            this.expTaskExecuteLogDaoRepo.save(etelm);
        }catch (Exception e){
            etelm=null;
            log.info("startTaskExeucte error");

        }
        return  etelm;
    }

    @Override
    @Async("asyncTask2")
    public void executeTask(ExpTaskModel etm,ExpTaskExecuteLogModel etelm) {
        //查询规则
        try {
            ExpRuleModel erm = this.expRuleDaoRepo.findByRuleId(etm.getRuleId());
            //查询存储模式
            DataStoreInfoModel dsim = this.dataStoreInfoDaoRepo.findByStoreId(etm.getStoreId());
            DataSourceInfoModel dsoim = dataSourceInfoDaoRepo
                .findByDataSourceId(etm.getDataSourceId());
            //查询任务下，规则详细
            List<ExpRuleDetailModel> erdmlist = expRuleDetailDaoRepo.findByRuleId(etm.getRuleId());
            if (erdmlist.size() > 0) {
                DataSource ds = DBMetaDataConfig.dataSourceMap.get(etm.getDataSourceId());
                for (int i = 0; i < erdmlist.size(); i++) {

                    ExpRuleDetailModel erdm = erdmlist.get(i);

                    //创建单表执行记录
                    //多线程执行单表
                    if(erdm.getSplitKey()!=null && !erdm.getSplitKey().equals("")){
                        SplitKeyExecutor executor;
                        //精确分片
                        if (erdm.getIsAccurate() == SplitKeyExecutorEnums.ACCURATE_SPLIT_EXECUTOR.getCode()) {
                            executor = new SplitKeyAdapter(accurateSplit);
                        //粗略分片
                        } else if (erdm.getIsAccurate() == SplitKeyExecutorEnums.ROUGH_SPLIT_EXECUTOR.getCode()) {
                            executor = new SplitKeyAdapter(roughSplit);
                        //创建分片执行器异常
                        } else {
                            throw new GlobalException(ExpConstants.NULL_OBJECT);
                        }
                        //创建分片任务处理者
                        SplitKeyTaskHandler splitKeyTaskHandler = new SplitKeyTaskHandler(ds,
                                erdm, dsim, expTaskExecuteDetailLogDaoRepo,
                                etm, etelm, localStorePath, fetchSize, printOutSize, scheduler, expTaskDao,
                                expruleDetailService);
                        //执行器执行分片和导出
                        executor.execute(splitKeyTaskHandler);
                    }else{
                        TaskExecuteCallable taskExecuteCallable = new TaskExecuteCallable(ds,
                            expTaskExecuteLogDao,
                            expTaskExecuteDetailLogDao, erdm, erm, dsim, expTaskExecuteDetailLogDaoRepo,
                            etm, etelm, dsoim, localStorePath,fetchSize,printOutSize, dataDictionaryService);
                        asyncTaskExecute.submit(taskExecuteCallable);
                    }
                }
            }
           // return true;
        }catch (Exception e){
            //任务执行失败
            this.expTaskExecuteLogDao.updateExeuceStatus(etelm.getTaskExecuteLogId(),3,0,"","");
            log.error("startTaskExeucte ",e);
           // return false;
        }
    }

    @Override
    public boolean updateExeuceStatus(String taskExecuteId, int status,int sumCount,String verifyPath,String message) {
        return this.expTaskExecuteLogDao.updateExeuceStatus(taskExecuteId,status,sumCount,verifyPath,message);
    }

    @Override
    public List<ExpTaskExecuteLogModel> findByStatus(int status) {
        return this.expTaskExecuteLogDaoRepo.findByStatus(status);
    }

    @Override
    public boolean deleteByTaskExecutelogId(String taskExecutelogId) {
        this.expTaskExecuteLogDao.deleteByTaskExecutelogId(taskExecutelogId);
        this.expTaskExecuteDetailLogDao.deleteByTaskExecuteLogId(taskExecutelogId);
        return true;
    }
}
